package com.test;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rabbitmq.client.Address;
import com.rabbitmq.client.ConnectionFactory;

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;
import reactor.rabbitmq.OutboundMessage;
import reactor.rabbitmq.RabbitFlux;
import reactor.rabbitmq.Sender;
import reactor.rabbitmq.SenderOptions;

public class Producer {

	private Logger LOGGER = LoggerFactory.getLogger(Producer.class);

	@Test
    public void testProducer() throws InterruptedException {
        int count = 2;
        CountDownLatch latch = new CountDownLatch(count);

        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.useNio();
        connectionFactory.setVirtualHost("/sms");
        connectionFactory.setUsername("sms_gaea-servicer-provider");
        connectionFactory.setPassword("dGGXPlY6YO");

        SenderOptions senderOptions =  new SenderOptions()
                .connectionFactory(connectionFactory)
                .connectionSupplier(cf -> cf.newConnection(
                        new Address[] {new Address("10.216.4.233",5672), new Address("10.216.4.234",5672), new Address("10.216.4.236",5672)},
                        "reactive-sender"))
                .connectionSubscriptionScheduler(Schedulers.elastic());

        Sender sender = RabbitFlux.createSender(senderOptions);
        sender
        	.sendWithPublishConfirms(Flux
        								.range(1, count)
        								.map(i -> {
        									return new OutboundMessage("sms.e.alevel", "alevel", ("Message_" + i).getBytes());
        								})
        							)
	        .doOnError(e -> LOGGER.error("Send failed", e))
	        .subscribe(r -> {
	            if (r.isAck()) {
	                LOGGER.info("Message {} sent successfully", new String(r.getOutboundMessage().getBody()));
	                latch.countDown();
	            }
	        });

        latch.await(10, TimeUnit.SECONDS);
        sender.close();
    }

}
